GET /apod:回傳每日天文圖(Astronomy Picture of the Day)GET /mars-photo:回傳火星照片GET /neo:回傳近地天體(Near-Earth Objects)資訊此為簡單的 FastAPI 專案,正式的作業環境應該把 route 分離到不同的檔案中。此專案結構僅作demo用:
nasa-app-mcp-demo/
├── .env
├── README.md
├── main.py
├── test_basic.py
├── venv/
│   └── ... (virtual environment files)
fastapi>=0.104.0
uvicorn[standard]>=0.24.0
httpx>=0.25.0
python-dotenv>=1.0.0
fastapi-mcp>=0.4.0
# 建資料夾
mkdir -p nasa-fastapi-mcp-demo/app && cd nasa-fastapi-mcp-demo
# 建立虛擬環境
python3.12 -m venv .venv && source .venv/bin/activate
# 建 requirements.txt(貼上上面內容)
# 然後安裝
pip install -r requirements.txt
建立 .env.example:
NASA_API_KEY=your_nasa_api_key_here
複製一份:
cp .env.example .env
# 然後把 your_nasa_api_key_here 改成你自己的 Key
main.pyimport os
import httpx
from typing import Optional, List, Dict, Any
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
app = FastAPI(
    title="Simple NASA MCP Server",
    description="A simple FastAPI server providing NASA data through REST endpoints and MCP",
    version="1.0.0"
)
# NASA API configuration
NASA_API_KEY = os.getenv("NASA_API_KEY", "DEMO_KEY")
NASA_BASE_URL = "https://api.nasa.gov"
# HTTP client for NASA API calls
client = httpx.AsyncClient(timeout=30.0)
# Response models
class APODResponse(BaseModel):
    title: str
    explanation: str
    url: str
    date: str
    media_type: str
    hdurl: Optional[str] = None
    copyright: Optional[str] = None
class MarsPhotoResponse(BaseModel):
    id: int
    sol: int
    camera: Dict[str, Any]
    img_src: str
    earth_date: str
    rover: Dict[str, Any]
class MarsPhotosResponse(BaseModel):
    photos: List[MarsPhotoResponse]
class NEOResponse(BaseModel):
    id: str
    name: str
    estimated_diameter: Dict[str, Any]
    is_potentially_hazardous_asteroid: bool
    close_approach_data: List[Dict[str, Any]]
class NEOFeedResponse(BaseModel):
    element_count: int
    near_earth_objects: Dict[str, List[NEOResponse]]
@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "service": "simple-nasa-mcp"}
@app.get("/apod", response_model=APODResponse)
async def get_astronomy_picture_of_day(date: Optional[str] = Query(None, description="Date in YYYY-MM-DD format")):
    """Get NASA's Astronomy Picture of the Day"""
    try:
        params = {"api_key": NASA_API_KEY}
        if date:
            params["date"] = date
            
        response = await client.get(f"{NASA_BASE_URL}/planetary/apod", params=params)
        
        if response.status_code == 200:
            data = response.json()
            return APODResponse(**data)
        else:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"NASA API error: {response.text}"
            )
    except httpx.RequestError as e:
        raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@app.get("/mars-photos", response_model=MarsPhotosResponse)
async def get_mars_rover_photos(
    rover: str = Query(..., description="Rover name (curiosity, perseverance, opportunity, spirit)"),
    sol: int = Query(..., description="Martian sol (day) number"),
    camera: Optional[str] = Query(None, description="Camera name (FHAZ, RHAZ, MAST, NAVCAM, etc.)")
):
    """Get Mars rover photos"""
    try:
        params = {
            "api_key": NASA_API_KEY,
            "sol": sol
        }
        if camera:
            params["camera"] = camera
            
        response = await client.get(f"{NASA_BASE_URL}/mars-photos/api/v1/rovers/{rover}/photos", params=params)
        
        if response.status_code == 200:
            data = response.json()
            return MarsPhotosResponse(**data)
        else:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"NASA API error: {response.text}"
            )
    except httpx.RequestError as e:
        raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@app.get("/neo", response_model=NEOFeedResponse)
async def get_near_earth_objects(
    start_date: str = Query(..., description="Start date in YYYY-MM-DD format"),
    end_date: str = Query(..., description="End date in YYYY-MM-DD format")
):
    """Get Near Earth Objects (asteroids and comets)"""
    try:
        params = {
            "api_key": NASA_API_KEY,
            "start_date": start_date,
            "end_date": end_date
        }
        
        response = await client.get(f"{NASA_BASE_URL}/neo/rest/v1/feed", params=params)
        
        if response.status_code == 200:
            data = response.json()
            return NEOFeedResponse(**data)
        else:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"NASA API error: {response.text}"
            )
    except httpx.RequestError as e:
        raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@app.on_event("shutdown")
async def shutdown_event():
    """Clean up HTTP client on shutdown"""
    await client.aclose()
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
uvicorn main:app --reload
--reload 參數會在代碼變更時自動重啟伺服器python main.py
這會啟動 FastAPI 伺服器,並在 http://127.0.0.1:8000 提供 API。
http://127.0.0.1:8000/docs
# 健康檢查
curl "http://127.0.0.1:8000/health"
# 今日 APOD
curl "http://127.0.0.1:8000/apod"
# 指定日期 APOD
curl "http://127.0.0.1:8000/apod?date=2024-07-20&hd=true"
# 搜尋圖片
curl "http://127.0.0.1:8000/mars-photos?rover=curiosity&sol=1000&camera=FHAZ"
# 近地天體
curl "http://127.0.0.1:8000/neo?start_date=2024-07-01&end_date=2024-07-31"
.env
YYYY-MM-DD
下一篇我們會用 fastapi-mcp 直接將這些endpoint轉為MCP可發現的tools,讓 VS Code 可以直接 discover 與呼叫。